#ifndef __JQSORT_H_
#define __JQSORT_H_

#include "jobqueue.h"

const uint64 kBuiltinSortSize = 100000;

template <class T, class V, class C>
  V PickPivot(T begin, T end, const C * comp) {
  V a[11];
  for (int32 i=0; i<11; i++) {
    a[i] = *(begin + (rand() % (end-begin))); 
  }
  sort(a, a+11, *comp);
  return a[5];
}

// swap two arrays
template <class T> struct SwapTask : public JobQueue::Task {
 SwapTask(T a, T b, uint64 n) 
   :a_(a), b_(b), n_(n) {
    // cout << "Creating swap task " << a_ << " " << b_ << " " << n_ << endl;
  }
  T a_;
  T b_;
  uint64 n_;
  void Run() {
    for (uint64 i=0; i<n_; i++) {
      // CHECK(b_[i] < a_[i]);
      iter_swap((a_+i), (b_+i));
    }
  }
};

template <class C> 
struct LEQComp {
  template <class T> 
  bool operator() (const T &a, const T &b) const {
    return !((*comp_)(b, a));
  }
LEQComp(const C *comp) :comp_(comp) {};
  const C *comp_;      
};

template <class T, class V, class C>
  struct SplitTask : public JobQueue::Task {
  void Run() {
    // cout << begin_ << " " << end_ << endl;
    //for (uint64 i=0; i<10; i++) {     
    //  sort(begin_, end_, *comp_);
    //  continue;
    T h = end_;
    T l = begin_;
    while (l < h) {
      if ((*comp_)(V(*l), split_value_)) {
	l++;
	continue;
      }
      if (!(*comp_)(V(*(h-1)), split_value_)) {
	h--;
	continue;
      }
      iter_swap(l, (h-1));
    }
    CHECK(l == h);
    split_ = l;
    // }
  }
  void Init(T begin, T end, const C *comp, V split_value) {
    begin_ = begin;
    end_ = end;
    comp_ = comp;
    split_value_ = split_value;
  }
  T begin_;
  T end_;
  T split_;
  const C * comp_;
  V split_value_;
};

// splits the array in place such that for elements t in the first part of the
// array (*comp)(t, pivot) is true, and false starting at the return value.
// returns the split position.
template <class T, class V, class C> 
  T JQSplit(T begin, T end, V pivot, const C * comp, JobQueue *jq) {
  uint64 n = end - begin;
  uint64 num_tasks = max(min(n / kBuiltinSortSize, 10000ull), 2ull);
  
  vector<SplitTask<T, V, C> *> tasks;
  JobQueue::Job split_job(jq);
  for (uint64 i=0; i<num_tasks; i++) {
    SplitTask<T, V, C> * t = new SplitTask<T, V, C>;
    tasks.push_back(t);
    t->autodelete_ = false;
    t->Init(begin + ShardBegin(n, num_tasks, i), 
	    begin + ShardEnd(n, num_tasks, i),
	    comp, pivot);
    split_job.AddTask(t);
  }
  split_job.Wait();
  // exit(1);

  uint64 low_block = 0;
  uint64 high_block = num_tasks - 1;
  T low_pos = tasks[low_block]->split_;
  T high_pos = tasks[high_block]->split_;
  T final_split;
  JobQueue::Job swap_job(jq);
  while (1) {
    if (low_pos == tasks[low_block]->end_) {
      low_block++;
      if (low_block == high_block) {
	final_split = high_pos;
	break;
      }
      low_pos = tasks[low_block]->split_;
      continue;
    }
    if (high_pos == tasks[high_block]->begin_) {
      high_block--;
      if (low_block == high_block) {
	final_split = low_pos;
	break;
      }
      high_pos = tasks[high_block]->split_;
      continue;
    }
    int64 size = min(tasks[low_block]->end_ - low_pos, 
		     high_pos - tasks[high_block]->begin_);
    swap_job.AddTask(new SwapTask<T>(low_pos, high_pos - size, size));
    low_pos += size;
    high_pos -= size;
  }
  for (uint64 i=0; i<num_tasks; i++) delete tasks[i];
  swap_job.Wait();
  // for (T * run = begin; run != final_split; run++) CHECK((*comp)(*run, pivot));
  // for (T * run = final_split; run != end; run++) CHECK(!(*comp)(*run, pivot));
  return final_split;
};

template <class T, class V, class C>
  void JQSort(T begin, T end, const C * comp, JobQueue * jq);

template <class T, class V, class C> 
  struct SortTask : public JobQueue::Task {
  void Run() {
    JQSort<T, V, C>(begin_, end_, comp_, job_->queue_);
  }
 SortTask(T begin, T end, const C *comp) 
   :begin_(begin), end_(end), comp_(comp) {}  
  T begin_;
  T end_;
  const C * comp_;
};

// T is the data type, C is the comparator class
template <class T, class V, class C>
  void JQSort(T begin, T end, const C * comp, JobQueue * jq) {
  uint64 n = end - begin;
  if (n < kBuiltinSortSize) {
    sort(begin, end, *comp); return;
  }
  V pivot = PickPivot<T, V, C>(begin, end, comp);
  T split_position = JQSplit(begin, end, pivot, comp, jq);
  JobQueue::Job sort_job(jq);
  if (begin < split_position)
    sort_job.AddTask (new SortTask<T, V, C>(begin, split_position, comp));
  
  double fraction = (double)(split_position - begin) / (end-begin);
  if (fraction > 0.2) {
    sort_job.AddTask (new SortTask<T, V, C>(split_position, end, comp));    
  } else {
    LEQComp<C> leqcomp(comp);
    T end_pivot = JQSplit(split_position, end, pivot, &leqcomp, jq);
    if (end_pivot != end)
      sort_job.AddTask(new SortTask<T, V, C>(end_pivot, end, comp));
  }
  sort_job.Wait();
}

#endif
